package org.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

object TransformRDD03 {

  def main(args: Array[String]): Unit = {

    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark")
    val sparkContext = new SparkContext(sparkConf)
    sparkContext.setLogLevel("ERROR")


    //转换算子
    /*
     1.map:将处理的数据逐条进行映射转换，这里的转换可以是类型转换，也可以是值的转换
     需求 ：将列表中的 1 2 3 4 转换为 2 4 6 8
     */
    val rdd1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    val resRdd1: RDD[Int] = rdd1.map((_ * 2)) // 一个rdd转换后，会生成一个新的rdd
    resRdd1.collect().foreach(println)

    /*
    2. mapPartitions:以为分区为单位进行数据处理
     */
    val rdd2: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4),2)//【1，2】  【3，4】
    //需求：找到每个分区的最大值
    val resRdd2: RDD[Int] = rdd2.mapPartitions(item => {
      //item  : 【1，2】  【3，4】
      List(item.max).iterator
    })
    resRdd2.collect().foreach(println)

    /*
    3.flatMap: 先映射转换 再进行扁平化（将两个子列表/数组，合并成一个列表）

      map 和 foreach 都是遍历列表/RDD里面的元素
        map可以返回每一个元素
        foreach不可以返回一个元素
     */
    val rdd3: RDD[List[Int]] = sparkContext.makeRDD(List(List(1, 2), List(3, 4))) // ==> List(1,2,3,4)
    val resRdd3: RDD[Int] = rdd3.flatMap(item => {
      //item:List(1, 2)
      item
    })
    resRdd3.collect().foreach(print)
    //需求：以空格拆分单词
    val rdd4: RDD[String] = sparkContext.makeRDD(List("Hello Spark", "Hello Hadoop"))//==> List(Hello,Spark,Hello,Hadoop)
    //由数组切割为字符串 mkString(" ")  字符串切割为数组 split(" ")
    val resRdd4: RDD[String] = rdd4.flatMap(item => {
      //item :  "Hello Spark"    "Hello Hadoop"
      item.split(" ") //  List ( [Hello,Spark],  [Hello,Hadoop]) ==>flatten List(Hello,Spark,Hello,Hadoop)
    })
    resRdd4.collect().foreach(println )

    /*
      4.glom: 将同一个分区的数据直接转换相同类型的数组进行处理，分区保持不变
     */
    val rdd5: RDD[Int] = sparkContext.makeRDD((1 to 4).toList, 2)// 【1，2】 【3，4】
    val resRdd5: RDD[Array[Int]] = rdd5.glom()  // List( [1,2],[3,4] )
    resRdd5.collect().foreach( data => println( data.mkString(",") ))

    //需求：获得两个分区最大值的和
    //1.将带有两个分区rdd转成rdd带有两个数组的形式
    val resRdd6: RDD[Array[Int]] = rdd5.glom() //List( [1,2],[3,4] )
    //2.对带有两个数组的rdd进行转换，得到每个数组中的最大值
    val resRdd7: RDD[Int] = resRdd6.map(array => {
      array.max
    })  // 2 4
    //3.进行最大值的相加
    println( resRdd7.sum())

    /*
      5.groupBy:将数组根据指定规则进行分组，分区默认不变。但是数据会被打乱重新组合，我们将这样的操作称之为shuffle。
                极限情况下，数据可能被分在同一个分区中。
     */
    val rdd6: RDD[Int] = sparkContext.makeRDD((1 to 4).toList, 2) // 【1，2】 【3，4】
    //需求：将奇数分为一组 偶数分为一组
    val resRdd8: RDD[(Int, Iterable[Int])] = rdd6.groupBy(_ % 2)
    resRdd8.collect().foreach( println )
    //需求：按照单词的首字母进行分组
    val rdd7: RDD[String] = sparkContext.makeRDD( List("Hello", "Spark", "Hadoop", "HBase", "Scala", "Python") )
    val resRdd9: RDD[(Char, Iterable[String])] = rdd7.groupBy(item => {
      //item "Hello" "Spark"
      item.charAt(0)
      //item(0)
    })
    resRdd9.collect().foreach( println )

    /*
      6.filter:将数据根据指定规则进行过滤。符合规则[true]的数据进行保留，不符合规则[false]的数据舍弃
     */
    val rdd8: RDD[Int] = sparkContext.makeRDD((1 to 4).toList)
    //获得Rdd中的奇数
    val resRdd10: RDD[Int] = rdd8.filter(_ % 2 != 0)
    resRdd10.collect().foreach( println )

    /*
      7.distinct:数据去重
     */
    val rdd9: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4, 5, 1, 2, 3, 4))
        //去重可以重新分区，如果不填写则采用的是默认分区。
    val resRdd11: RDD[Int] = rdd9.distinct(3)
    println("分区数：",resRdd11.getNumPartitions)
    resRdd11.collect().foreach( println )

    /*
      8. sortBy: 默认是升序。如果想变成降序，第二个参数为false
     */
    val resRdd12: RDD[Int] = resRdd11.sortBy(num => num, false)
    resRdd12.collect().foreach( println )
    //需求：按照key进行排序，默认升序
    val rdd10: RDD[(String, String)] = sparkContext.makeRDD(List( ("1", "a"),  ("3", "c"), ("2", "b"), ("4", "d")  ))
    var resRdd13: RDD[(String, String)] = rdd10.sortBy(_._1)
    val resRdd14: RDD[(String, String)] = rdd10.sortByKey()


    /*
      9.repatition:
           底层coalesce操作（这个算子只能做减少分区的操作，减小分区可以减少资源分配），
           repatition不仅仅可以减少分区还是增加分区。同时可以尽量避免数据倾斜的问题

           【1，2】 【3，4】 【5，6】
           理想情况下：  【1，2，5】  【3，4，6】  【1，2，3】 【4，5，6】
           数据倾斜： 【1，2，3，4，5，6】 【 】     【1，2，3，4，5】 【6】
    */
    val rdd11: RDD[Int] = sparkContext.makeRDD((1 to 6).toList,3)
    //rdd11.saveAsTextFile("output4")
    val resRdd15: RDD[Int] = rdd11.repartition(2)
    //resRdd15.saveAsTextFile("output5")
    val resRdd16: RDD[Int]  =  rdd11.repartition(5)
    //resRdd16.saveAsTextFile("output6")

    /*
      10.交集 并集 差集  拉链

     */
    val rdd12: RDD[Int] = sparkContext.makeRDD((1 to 4).toList) //1 2 3 4
    val rdd13: RDD[Int] = sparkContext.makeRDD((3 to 6).toList) //3 4 5 6

    //交集：
    val resRdd17: RDD[Int] = rdd12.intersection(rdd13)
    println(resRdd17.collect().mkString(","))

    //并集
    val resRdd18: RDD[Int] = rdd12.union(rdd13)
    println(resRdd18.collect.mkString(","))

    //差集                   rdd12 差 rdd13 那么结果就是rdd12的差集 1，2
    val resRdd19: RDD[Int] = rdd12.subtract(rdd13)
    println(resRdd19.collect.mkString(","))

    //拉链
    val resRdd20: RDD[(Int, Int)] = rdd12.zip(rdd13) // [  (1,3),(2,4),(3,5),(4,6) ]
    println(resRdd20.collect().mkString(","))


    /*
      11.partitionBy:将数据按照指定Partitioner重新进行分区，Spark默认的分区器HashPartitioner
     */
    val rdd14: RDD[Int] = sparkContext.makeRDD((1 to 4).toList, 2)
    //隐式转换
    val resRdd21: RDD[(Int, Int)] = rdd14.map((_, 1)) // [  【(1,1),(2,1)】,  【(3,1),(4,1)】 ]
    //resRdd21.saveAsTextFile("output7")
    val resRdd22: RDD[(Int, Int)] = resRdd21.partitionBy(new HashPartitioner(3))
    //resRdd22.saveAsTextFile("output9")

    /*
      12.reduceByKey:可以将数据按照相同Key对数据进行聚合
     */
    val rdd15: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("b", 4), ("b", 5)))
    /*
          a  x =1 y =2  1+2 =3
          b  x=4  y=5   4+5 =9
     */
    var resRdd23: RDD[(String, Int)] = rdd15.reduceByKey((x: Int, y: Int) => {
      x + y
    })
    resRdd23 = rdd15.reduceByKey(_+_)
    resRdd23.collect().foreach( println )

    /*
     13.groupByKety:将数据根据Key对Value进行分组
     */               // Key    对应值的迭代器 1 2
    val resRdd24: RDD[(String, Iterable[Int])] = rdd15.groupByKey()
    resRdd24.collect().foreach( println )
    /*
    分组: 将数组/集合按照指定条件进行分组
     需求： 1. 按照性别进行分组
           2. 统计不同性别的人数
     */
    val rdd16: RDD[(String, String)] = sparkContext.makeRDD(List( ("男", "张三"), ("女", "李四"), ("男", "王五")))
    //println( rdd16.countByKey())
                      //男，    [张三，王五]
    val resRdd25: RDD[(String, Iterable[String])] = rdd16.groupByKey()
    val resRdd26: RDD[(String, Int)] = resRdd25.map(x => (x._1, x._2.size))
    resRdd26.collect().foreach( println )

    /*
      14.aggregateByKey:将数据根据 不同的规则 进行分区内计算和分区间计算
                  ByKey:根据Key进行操作
          1分区： 【 (a,1),(a,2),(b,3), (a,0),(b,0)  】 ==> 根据Key找到该分区的最大值  (a,2)  (b,3)
          2分区： 【(b,4 ),(b,5),(a,6), (a,0),(b,0) 】   ==> 根据Key找到该分区的最大值 (b,5)   (a,6)

          分区间的最大值的和
            a  2 + 6   (a,8)
            b  3 + 5   (b,8)
     */

    val rdd17: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("b", 3),
                                                              ("b", 4), ("b", 5), ("a", 6)
                                                                                  ), 2)
    val resRdd27: RDD[(String, Int)] = rdd17.aggregateByKey(0)(
      (x, y) => math.max(x, y), //根据Key找到该分区的最大值
      (x, y) => x + y  //分区间的操作
    )
    resRdd27.collect().foreach( println )

    //需求：计算两个分区间的 key的value的总和
    var resRdd28: RDD[(String, Int)] = rdd17.aggregateByKey(0)(
      //分区内操作:求分区内的和
      /*
         1. key :a  1+2=3   b 3
         2  key a:  6     b 4 + 5=9
       */
      (x, y) => x + y,
      //分区间操作：求分区间的和
      //1 a:3  b:3
      //2:a:6  b:9
      // a: 3 + 6  b: 3 + 9
      (x, y) => x + y
    )
    //简写方式
    resRdd28 = rdd17.aggregateByKey(0)(_+_,_+_)
    //当分区内和分区间的计算规则相同的时候，可以使用flodByKey
    resRdd28 = rdd17.foldByKey(0)(_+_)
    resRdd28.collect().foreach( println )

    //需求：使用aggregateByKey 计算rdd17的每个键出现次数以及每个Key的value总和。
    // (0,0) 第一个0是用来存Key出现的次数  第二个0用来存每个Key的值的总和        出现的次数     值
   /* 1分区： 【 (a,1),(a,2),(b,3), (a,(0,0)) ) ,(b,(0,0) ) 】 ==> a: (0,0) + (1,            1)==(1,1) +  (1,2) ==> a: (2,3)
                                                              ==> b: (0,0) + (1,            3) ==> b: (1,3)
      2分区： 【(b,4 ),(b,5),(a,6), (a,(0,0)) ) ,(b,(0,0) ) 】 ==> a: (0,0)+  (1,            6) ==> a:(1,6)
                                                              ==> b: (0,0)+  (1,            4) ==> (1,4) +(1,5)==> b:(2,9)
      分区内的操作结果   第一分区 (a,(2,3 ) )  (b,(1,3))

                        第二分区 (a,(1,6 ) )  (b,(2,9))

      分区间操作：      (a,(2,3 ) )  + (a,(1,6 ) ) = (a,(3,9))      (b,(1,3))   +     (b,(2,9)) =(b,(3,12))
    */
    //                 (a,       ( 3,  9)  )
    val resRdd29: RDD[(String, (Int, Int))] = rdd17.aggregateByKey((0, 0))(
      /*分区内操作
           a出现的次数: 0+1 + 1,a的值总和 0+1+2
           b出现的次数： 0+1  , b的值的总和 0+3
       */
      (t, v) => { //t就是 (0,0) v就是 元素中的value
        (t._1 + 1, t._2 + v)
      },
      (t1, t2) => { //t1 表示 第一个分区的数据 t2 第二个分区的数据
        (t1._1 + t2._1, t1._2 + t2._2) //( 3 ,9)  (3,12)
      }
    )
    resRdd29.collect().foreach( println )
    //平均值             (a,3) (b,4)
    var resRdd30: RDD[(String, Int)] = resRdd29.mapValues(t => { //t 就是表示 元组  (3,9)  (3,12)
      t._2 / t._1
    })
    // (a,(3,9))    (b,(3,12))
    resRdd30 = resRdd29.mapValues{
      case (count,total)=>{
        total / count
      }
    }
    resRdd30.collect().foreach( println )

    //15.sortByKey: 根据Key来进行排序的
    val rdd18: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("c", 2), ("b", 3)))
    val resRdd31: RDD[(String, Int)] = rdd18.sortByKey()//默认为true 升序排序 false降序排序
    resRdd31.collect().foreach( println )

    /*
     join leftOuterJoin rightOuterJoin
     */
    val rdd19: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("c", 3),("b", 8)))
    val rdd20: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 5), ("c",6), ("a", 3),("g", 8)))
    val resRdd32: RDD[(String, (Int, Int))] = rdd19.join(rdd20)
    resRdd32.collect().foreach( println )
/*
      (a,(1,5))
      (a,(1,3))
      (a,(2,5))
      (a,(2,3))
      (c,(3,6))
 */
    val resRdd33: RDD[(String, (Int, Option[Int]))] = rdd19.leftOuterJoin(rdd20)
    resRdd33.collect().foreach( println )
    val resRdd34: RDD[(String, (Option[Int], Int))] = rdd19.rightOuterJoin(rdd20)
    resRdd34.collect().foreach( println )

    /*
    16.cogroup: 分组后 拼接  （connect + group ）。分组拼接后的结果 同一个Rdd的值会放到同一个序列下
     */
    val resRdd35: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd19.cogroup(rdd20)
    resRdd35.collect().foreach( println )
    /*
    (a,(Seq(1, 2),Seq(5, 3)))
    (b,(Seq(8),Seq()))
    (c,(Seq(3),Seq(6)))
    (g,(Seq(),Seq(8)))
     */

    sparkContext.stop()
  }
}
